#include "stdafx.h"
#include "ORB_Notify_Consumer.h"
#include "tao/debug.h"
#include "ace/High_Res_Timer.h"
#include "ace/Stats.h"
#include "ace/Throughput_Stats.h"
#include "ace/OS_NS_stdio.h"
#include "ace/OS_NS_unistd.h"

ORB_Notify_Consumer::ORB_Notify_Consumer( void ) :
proxy_supplier_thread_count_ (0)
{

}

ORB_Notify_Consumer::~ORB_Notify_Consumer (void)
{
    //// Disconnect from the EC
    //this->disconnect ();

    //// Deactivate this object.
    //this->deactivate ();
}

void
ORB_Notify_Consumer::init_consumer (PortableServer::POA_var& poa,
                                    CosNotifyChannelAdmin::EventChannel_var& event_channel,
                                    CosNotifyChannelAdmin::ConsumerAdmin_var& consumer_admin,
                                    int proxy_supplier_thread_count)
{
    this->default_POA_ = poa;
    this->event_channel_ = event_channel;
    this->consumer_admin_ = consumer_admin;
    this->proxy_supplier_thread_count_ = proxy_supplier_thread_count;

    this->connect ();
    this->subscribe_events();
    this->add_filters();
}

void
ORB_Notify_Consumer::connect (void)
{
    // Activate the consumer with the default_POA_
    CosNotifyChannelAdmin::ProxySupplier_var proxy_supplier;
    CosNotifyChannelAdmin::ProxyID proxy_supplier_id_;
    
    if (this->proxy_supplier_thread_count_ != 0){
        // Narrow to the extended interface.
        NotifyExt::ConsumerAdmin_var admin_ext = NotifyExt::ConsumerAdmin::_narrow (this->consumer_admin_.in ());
        NotifyExt::ThreadPoolParams tp_params = {
            NotifyExt::CLIENT_PROPAGATED,       // NotifyExt::PriorityModel priority_model;
            0,                                  // NotifyExt::Priority      server_priority;
            0,                                  // ::CORBA::ULong           stacksize;
            this->proxy_supplier_thread_count_, // ::CORBA::ULong           static_threads;
            0,                                  // ::CORBA::ULong           dynamic_threads;
            0,                                  // NotifyExt::Priority      default_priority;
            0,                                  // ::CORBA::Boolean         allow_request_buffering;
            0,                                  // ::CORBA::ULong           max_buffered_requests;
            0                                   // ::CORBA::ULong           max_request_buffer_size;
        };

        CosNotification::QoSProperties qos (1);
        qos.length (1);
        qos[0].name = CORBA::string_dup (NotifyExt::ThreadPool);
        qos[0].value <<= tp_params;

        // Obtain the proxy. The QoS is applied to the POA in which the Proxy is hosted.
        proxy_supplier = admin_ext->obtain_notification_push_supplier_with_qos
            (CosNotifyChannelAdmin::STRUCTURED_EVENT, proxy_supplier_id_, qos);
    }
    else{
        proxy_supplier = this->consumer_admin_->obtain_notification_push_supplier
            (CosNotifyChannelAdmin::STRUCTURED_EVENT, proxy_supplier_id_);
    }

    ACE_ASSERT (!CORBA::is_nil (proxy_supplier.in ()));

    // narrow
    this->proxy_supplier_ = CosNotifyChannelAdmin::StructuredProxyPushSupplier::_narrow (proxy_supplier.in ());
    
    ACE_ASSERT (!CORBA::is_nil (proxy_supplier_.in ()));
    
    this->proxy_supplier_->connect_structured_push_consumer(this->_this());

    ACE_DEBUG ((LM_DEBUG, ACE_TEXT("Created ORB_Notify_Consumer %d with %d threads.\n"),
        proxy_supplier_id_, this->proxy_supplier_thread_count_));
}

void
ORB_Notify_Consumer::disconnect (void)
{
	if( this->proxy_supplier_ )
		this->proxy_supplier_->disconnect_structured_push_supplier();
}

void
ORB_Notify_Consumer::deactivate (void)
{
    PortableServer::POA_var poa (this->_default_POA ());
    PortableServer::ObjectId_var id (poa->servant_to_id (this));
    poa->deactivate_object (id.in());
}

void ORB_Notify_Consumer::subscribe_event(const char * domain_name, const char * type_name)
{
    CosNotification::EventTypeSeq added (1);
    CosNotification::EventTypeSeq removed;

    added.length (1);
    added[0].domain_name = CORBA::string_dup (domain_name);
    added[0].type_name   = CORBA::string_dup (type_name);

    // Call subscription_change to inform the supplier that this consumer is available.
    this->proxy_supplier_->subscription_change (added, removed);

    ACE_DEBUG((LM_DEBUG,ACE_TEXT("%I(%P|%t) subscribe_event %C %C\n"),domain_name,type_name));
}
#pragma region // = ServantBase operations

PortableServer::POA_ptr
ORB_Notify_Consumer::_default_POA (void)
{
    return PortableServer::POA::_duplicate (this->default_POA_.in ());
}

#pragma endregion

#pragma region // = NotifyPublish method

void
ORB_Notify_Consumer::offer_change (const CosNotification::EventTypeSeq & /*added*/,
                                   const CosNotification::EventTypeSeq & /*removed*/)
{
    // No-Op.
}

#pragma endregion

#pragma region // = StructuredPushSupplier methods

void
ORB_Notify_Consumer::push_structured_event (const CosNotification::StructuredEvent & event)
{
    ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
    event_sig_(event,m_name);
}

void
ORB_Notify_Consumer::disconnect_structured_push_consumer (void)
{
    this->deactivate ();
}

#pragma endregion

